package org.study.hadoop.md.ch4;

import org.apache.hadoop.conf.Configurable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.study.hadoop.md.util.MRDPUtils;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Map;

/**
 * 分区：给定一组用户信息，按照最近访问日期中的年份信息对记录进行分区，一年对应一个分区
 */
public class PartitionedUsers extends Configured implements Tool {

    public static class ValueReducer extends Reducer<IntWritable, Text, Text, NullWritable> {
        @Override
        protected void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            for (Text text : values) {
                context.write(text, NullWritable.get());
            }
        }
    }

    public static class LastAccessDatePartitioner extends Partitioner<IntWritable, Text> implements Configurable {
        private static final String MIN_LAST_ACCESS_DATE_YEAR = "min.last.access.date.year";
        private Configuration conf = null;
        private int minLastAccessDateYear = 0;

        @Override
        public void setConf(Configuration conf) {
            this.conf = conf;
            minLastAccessDateYear = conf.getInt(MIN_LAST_ACCESS_DATE_YEAR, 0);
        }

        @Override
        public Configuration getConf() {
            return conf;
        }

        @Override
        public int getPartition(IntWritable key, Text value, int numPartitions) {
            return key.get() - minLastAccessDateYear;
        }

        public static void setMinLastAccessDateYear(Job job, int minLastAccessDateYear) {
            job.getConfiguration().setInt(MIN_LAST_ACCESS_DATE_YEAR, minLastAccessDateYear);
        }
    }

    public static class LastAccessDateMapper extends Mapper<Object, Text, IntWritable, Text> {
        private final static SimpleDateFormat FRMT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        private IntWritable outKey = new IntWritable();

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
            String strDate = parsed.get("LastAccessDate");
            if (strDate != null) {
                try {
                    Calendar calendar = Calendar.getInstance();
                    calendar.setTime(FRMT.parse(strDate));
                    outKey.set(calendar.get(Calendar.YEAR));
                    context.write(outKey, value);
                } catch (ParseException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Path outPath = new Path("out");
        Path inPath = new Path("user.txt");
        FileSystem fs = outPath.getFileSystem(getConf());
        if (fs.exists(outPath)) {
            fs.delete(outPath, true);
        }
        Job job = Job.getInstance(getConf(), "PostCommentHierarchy");
        job.setJarByClass(PartitionedUsers.class);
        job.setMapperClass(LastAccessDateMapper.class);
        job.setPartitionerClass(LastAccessDatePartitioner.class);
        LastAccessDatePartitioner.setMinLastAccessDateYear(job, 2011);
        job.setNumReduceTasks(6);
        job.setReducerClass(ValueReducer.class);
        FileInputFormat.addInputPath(job, inPath);
        FileOutputFormat.setOutputPath(job, outPath);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        job.getConfiguration().set("mapred.textoutputformat.separator", "");
        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new PartitionedUsers(), args);
        System.exit(res);
    }
}
